package org.myfram.flink.flinkonjar.common.connector.hbase.source;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.hadoop.conf.Configuration;
import org.myfram.flink.flinkonjar.common.config.HBaseConfig;
import org.myfram.flink.flinkonjar.common.connector.hbase.HBaseRowConvert;
import org.myfram.flink.flinkonjar.common.properties.HBaseProperties;

import java.util.ArrayList;
import java.util.List;

public class HBaseSourceBuilder<T> {

    private String tableName;
    private HBaseRowConvert<T> hbaseRowConvert;
    private Configuration hbaseConfig;

    private List<String> columnFamilies = new ArrayList<>();
    private List<Tuple2<String, String>> columns = new ArrayList<>();
    private int cacheSize = 1000;
    private long timeRangeStart = 0L;
    private long timeRangeEnd = Long.MAX_VALUE;

    public HBaseSourceBuilder(HBaseProperties prop) {
        this.hbaseConfig = HBaseConfig.getSourceConf(prop);
    }

    public  HBaseSourceBuilder<T> setTimeRange(long timeRangeStart, long timeRangeEnd) {
        this.timeRangeStart = timeRangeStart;
        this.timeRangeEnd = timeRangeEnd;
        return this;
    }

    public  HBaseSourceBuilder<T> setTableName(String tableName) {
        this.tableName = tableName;
        return this;
    }

    public HBaseSourceBuilder<T> setHbaseRowConvert(HBaseRowConvert<T> hbaseRowConvert) {
        this.hbaseRowConvert = hbaseRowConvert;
        return this;
    }

    public HBaseSourceBuilder<T> addColumnFamily(String columnFamily) {
        this.columnFamilies.add(columnFamily);
        return this;
    }

    public HBaseSourceBuilder<T> addColumn(String family, String column) {
        this.columns.add(new Tuple2<>(family, column));
        return this;
    }

    public HBaseSourceBuilder<T> setCacheSize(int cacheSize) {
        this.cacheSize = cacheSize;
        return this;
    }

    public SourceFunction<T> build() {
        return new HBaseSource<>(tableName,
                hbaseRowConvert,
                hbaseConfig,
                columnFamilies,
                columns,
                cacheSize,
                timeRangeStart,
                timeRangeEnd);
    }

}
